package org.lds.message.center.plugin.core.kafka;

import lombok.RequiredArgsConstructor;
import org.apache.commons.collections4.CollectionUtils;
import org.apache.kafka.clients.consumer.ConsumerRecord;
import org.lds.message.center.common.util.JsonUtil;
import org.springframework.kafka.support.Acknowledgment;

import java.util.List;
import java.util.Objects;
import java.util.stream.Collectors;

@RequiredArgsConstructor
public class AbstractKafkaConsumer<T> implements KafkaConsumer {
    private final Class<T> clazz;

    @Override
    public void consume(List<ConsumerRecord<String, String>> consumerRecords, Acknowledgment ack) {
        List<T> records = getRecords(consumerRecords);
        if (CollectionUtils.isNotEmpty(records)) {
            doConsume(records);
        }

        ack.acknowledge();
    }

    private List<T> getRecords(List<ConsumerRecord<String, String>> records) {
        return records.stream()
                .map(a -> JsonUtil.fromJson(a.value(), clazz))
                .filter(a -> Objects.nonNull(a))
                .collect(Collectors.toList());
    }

    protected void doConsume(List<T> records) {
    }
}
